/*-------------------------------------------------------------------------
 *
 * nodeGather.c
 *	  支持通过多个工作线程扫描计划的例程。
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * Gather 执行器启动并行工作线程来运行多个计划副本。 如果工作线程不可用或尚未启动，它也可以运行计划本身。 然后，它将所有生成的结果与工作线程的结果合并为一个输出流。 因此，通常会与运行同一计划多个副本不会产生重复输出的计划一起使用，例如并行感知的 SeqScan。
 *
 * 或者，可以配置 Gather 节点仅使用一个工作线程，并且可以设置单副本标志。在这种情况下，Gather 节点将在一个工作线程中运行计划，而不会执行该计划本身。 在这种情况下，它只返回工作线程返回的任何元组。 如果无法获得工作线程，则会运行计划本身并返回结果。因此，用于单副本 Gather 节点的计划不需要并行感知。
 *
 * IDENTIFICATION
 *	  src/backend/executor/nodeGather.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/relscan.h"
#include "access/xact.h"
#include "executor/execdebug.h"
#include "executor/execParallel.h"
#include "executor/nodeGather.h"
#include "executor/nodeSubplan.h"
#include "executor/tqueue.h"
#include "miscadmin.h"
#include "optimizer/optimizer.h"
#include "pgstat.h"
#include "utils/memutils.h"
#include "utils/rel.h"


static TupleTableSlot *fc_ExecGather(PlanState *fc_pstate);
static TupleTableSlot *fc_gather_getnext(GatherState *fc_gatherstate);
static MinimalTuple fc_gather_readnext(GatherState *fc_gatherstate);
static void fc_ExecShutdownGatherWorkers(GatherState *fc_node);


/* ----------------------------------------------------------------
 *		ExecInitGather
 * ----------------------------------------------------------------
 */
GatherState *
ExecInitGather(Gather *fc_node, EState *fc_estate, int fc_eflags)
{
	GatherState *fc_gatherstate;
	Plan	   *fc_outerNode;
	TupleDesc	fc_tupDesc;

	
/* Gather节点没有innerPlan节点。 */
	Assert(innerPlan(fc_node) == NULL);

	/*
	 * 创建状态结构
	 */
	fc_gatherstate = makeNode(GatherState);
	fc_gatherstate->ps.plan = (Plan *) fc_node;
	fc_gatherstate->ps.state = fc_estate;
	fc_gatherstate->ps.ExecProcNode = fc_ExecGather;

	fc_gatherstate->initialized = false;
	fc_gatherstate->need_to_scan_locally =
		!fc_node->single_copy && parallel_leader_participation;
	fc_gatherstate->tuples_needed = -1;

	/*
	 * 杂项初始化
	 *
	 * 为节点创建表达式上下文
	 */
	ExecAssignExprContext(fc_estate, &fc_gatherstate->ps);

	/*
	 * 现在初始化外部计划
	 */
	fc_outerNode = outerPlan(fc_node);
	outerPlanState(fc_gatherstate) = ExecInitNode(fc_outerNode, fc_estate, fc_eflags);
	fc_tupDesc = ExecGetResultType(outerPlanState(fc_gatherstate));

	/*
	 * 领导者可能直接访问 ExecProcNode 结果（如果需要本地扫描），
	 * 或通过元组队列从工作进程获取。因此，我们不能
	 * 简单地依赖于在此节点内评估的表达式的槽类型是固定的。
	 */
	fc_gatherstate->ps.outeropsset = true;
	fc_gatherstate->ps.outeropsfixed = false;

	/*
	 * 初始化结果类型和投影。
	 */
	ExecInitResultTypeTL(&fc_gatherstate->ps);
	ExecConditionalAssignProjectionInfo(&fc_gatherstate->ps, fc_tupDesc, OUTER_VAR);

	/*
	 * 在没有投影的情况下，结果槽类型并不是简单已知的，见上面的评论。
	 */
	if (fc_gatherstate->ps.ps_ProjInfo == NULL)
	{
		fc_gatherstate->ps.resultopsset = true;
		fc_gatherstate->ps.resultopsfixed = false;
	}

	/*
	 * 将漏斗槽初始化为与外部计划相同的元组描述符。
	 */
	fc_gatherstate->funnel_slot = ExecInitExtraTupleSlot(fc_estate, fc_tupDesc,
													  &TTSOpsMinimalTuple);

	/*
	 * Gather 不支持检查条件（在子节点执行总是更有效）。
	 */
	Assert(!fc_node->plan.qual);

	return fc_gatherstate;
}

/* ----------------------------------------------------------------
 *		ExecGather(node)
 *
 *		通过多个工作进程扫描关系并返回
 *		下一个符合条件的元组。
 * ----------------------------------------------------------------
 */
static TupleTableSlot * fc_ExecGather(PlanState *fc_pstate)
{
	GatherState *fc_node = castNode(GatherState, fc_pstate);
	TupleTableSlot *fc_slot;
	ExprContext *fc_econtext;

	CHECK_FOR_INTERRUPTS();

	/*
	 * 在第一次执行时初始化并行上下文和工作进程。我们在第一次执行时
	 * 而不是在节点初始化期间进行此操作，因为它需要分配一个大型动态段，
	 * 所以最好仅在真正需要时进行。
	 */
	if (!fc_node->initialized)
	{
		EState	   *fc_estate = fc_node->ps.state;
		Gather	   *fc_gather = (Gather *) fc_node->ps.plan;

		/*
		 * 有时我们可能不得不在没有并行处理的情况下运行；但如果并行
		 * 模式处于活动状态，那么我们可以尝试启动一些工作进程。
		 */
		if (fc_gather->num_workers > 0 && fc_estate->es_use_parallel_mode)
		{
			ParallelContext *fc_pcxt;

			/* 初始化或重新初始化工作进程所需的共享状态。 */
			if (!fc_node->pei)
				fc_node->pei = ExecInitParallelPlan(fc_node->ps.lefttree,
												 fc_estate,
												 fc_gather->initParam,
												 fc_gather->num_workers,
												 fc_node->tuples_needed);
			else
				ExecParallelReinitialize(fc_node->ps.lefttree,
										 fc_node->pei,
										 fc_gather->initParam);

			/*
			 * 注册后端工作进程。我们可能不会得到和请求一样多的工作进程，
			 * 或者根本没有。
			 */
			fc_pcxt = fc_node->pei->pcxt;
			LaunchParallelWorkers(fc_pcxt);
			/* 我们保存启动的工作进程数量，以便 EXPLAIN 受益 */
			fc_node->nworkers_launched = fc_pcxt->nworkers_launched;

			/* 设置元组队列读取器以读取结果。 */
			if (fc_pcxt->nworkers_launched > 0)
			{
				ExecParallelCreateReaders(fc_node->pei);
				/* 制作一个工作数组，显示活动的读取器 */
				fc_node->nreaders = fc_pcxt->nworkers_launched;
				fc_node->reader = (TupleQueueReader **)
					palloc(fc_node->nreaders * sizeof(TupleQueueReader *));
				memcpy(fc_node->reader, fc_node->pei->reader,
					   fc_node->nreaders * sizeof(TupleQueueReader *));
			}
			else
			{
				/* 没有工作进程？那就算了。 */
				fc_node->nreaders = 0;
				fc_node->reader = NULL;
			}
			fc_node->nextreader = 0;
		}

		/* 如果没有工作进程或已启用且不是单一副本，则本地运行计划。 */
		fc_node->need_to_scan_locally = (fc_node->nreaders == 0)
			|| (!fc_gather->single_copy && parallel_leader_participation);
		fc_node->initialized = true;
	}

	/*
	 * 重置每个元组的内存上下文，以释放在上一个元组周期中分配的任何表达式评估
	 * 存储。
	 */
	fc_econtext = fc_node->ps.ps_ExprContext;
	ResetExprContext(fc_econtext);

	/*
	 * 获取下一个元组，可能来自我们的一个工作进程，或者通过自己运行计划
	 * 来获取。
	 */
	fc_slot = fc_gather_getnext(fc_node);
	if (TupIsNull(fc_slot))
		return NULL;

	/* 如果不需要投影，我们就完成了。 */
	if (fc_node->ps.ps_ProjInfo == NULL)
		return fc_slot;

	/*
	 * 使用 ExecProject() 形成结果元组，并返回它。
	 */
	fc_econtext->ecxt_outertuple = fc_slot;
	return ExecProject(fc_node->ps.ps_ProjInfo);
}

/* ----------------------------------------------------------------
 *		ExecEndGather
 *
 *		释放通过 C 例程分配的任何存储。
 * ----------------------------------------------------------------
 */
void ExecEndGather(GatherState *fc_node)
{
	ExecEndNode(outerPlanState(fc_node));	/* 让子进程先清理 */
	ExecShutdownGather(fc_node);
	ExecFreeExprContext(&fc_node->ps);
	if (fc_node->ps.ps_ResultTupleSlot)
		ExecClearTuple(fc_node->ps.ps_ResultTupleSlot);
}

/*
 * 读取下一个元组。我们可能会使用 gather_readnext 从一个元组队列中获取元组，
 * 或者如果没有元组队列包含元组且未设置 single_copy 标志，我们可能会
 * 在本地生成一个元组。
 */
static TupleTableSlot * fc_gather_getnext(GatherState *fc_gatherstate)
{
	PlanState  *fc_outerPlan = outerPlanState(fc_gatherstate);
	TupleTableSlot *fc_outerTupleSlot;
	TupleTableSlot *fc_fslot = fc_gatherstate->funnel_slot;
	MinimalTuple fc_tup;

	while (fc_gatherstate->nreaders > 0 || fc_gatherstate->need_to_scan_locally)
	{
		CHECK_FOR_INTERRUPTS();

		if (fc_gatherstate->nreaders > 0)
		{
			fc_tup = fc_gather_readnext(fc_gatherstate);

			if (HeapTupleIsValid(fc_tup))
			{
				ExecStoreMinimalTuple(fc_tup,	/* 存储的元组 */
									  fc_fslot,	/* 存储元组的槽 */
									  false);	/* 不要释放元组  */
				return fc_fslot;
			}
		}

		if (fc_gatherstate->need_to_scan_locally)
		{
			EState	   *fc_estate = fc_gatherstate->ps.state;

			/* 在执行计划时安装我们的 DSA 区域。 */
			fc_estate->es_query_dsa =
				fc_gatherstate->pei ? fc_gatherstate->pei->area : NULL;
			fc_outerTupleSlot = ExecProcNode(fc_outerPlan);
			fc_estate->es_query_dsa = NULL;

			if (!TupIsNull(fc_outerTupleSlot))
				return fc_outerTupleSlot;

			fc_gatherstate->need_to_scan_locally = false;
		}
	}

	return ExecClearTuple(fc_fslot);
}

/*
 * 尝试从我们的一位并行工作进程读取一个元组。
 */
static MinimalTuple fc_gather_readnext(GatherState *fc_gatherstate)
{
	int			fc_nvisited = 0;

	for (;;)
	{
		TupleQueueReader *fc_reader;
		MinimalTuple fc_tup;
		bool		fc_readerdone;

		/* 检查异步事件，特别是来自工作进程的消息。 */
		CHECK_FOR_INTERRUPTS();

		/*
		 * 尝试读取一个元组，但如果没有可用的，则不阻塞。
		 *
		 * 注意，TupleQueueReaderNext 对于无法初始化的工作进程将
		 * 返回 NULL。我们将视该工作进程为未生成任何元组；
		 * 当我们到达那里时，WaitForParallelWorkersToFinish 将出错。
		 */
		Assert(fc_gatherstate->nextreader < fc_gatherstate->nreaders);
		fc_reader = fc_gatherstate->reader[fc_gatherstate->nextreader];
		fc_tup = TupleQueueReaderNext(fc_reader, true, &fc_readerdone);

		/*
		 * 如果这个读取器完成了，将其从我们活动的工作数组中移除。
		 * 如果所有读取器都完成了，我们就离开这里。
		 */
		if (fc_readerdone)
		{
			Assert(!fc_tup);
			--fc_gatherstate->nreaders;
			if (fc_gatherstate->nreaders == 0)
			{
				fc_ExecShutdownGatherWorkers(fc_gatherstate);
				return NULL;
			}
			memmove(&fc_gatherstate->reader[fc_gatherstate->nextreader],
					&fc_gatherstate->reader[fc_gatherstate->nextreader + 1],
					sizeof(TupleQueueReader *)
					* (fc_gatherstate->nreaders - fc_gatherstate->nextreader));
			if (fc_gatherstate->nextreader >= fc_gatherstate->nreaders)
				fc_gatherstate->nextreader = 0;
			continue;
		}

		/* 如果我们得到了一个元组，就返回它。 */
		if (fc_tup)
			return fc_tup;

		/*
		 * 以轮询方式推进 nextreader 指针。注意，只有在我们不能从
		 * 当前工作进程获取元组时，我们才会到达这段代码。
		 * 我们过去是在每个元组之后推进 nextreader 指针，但结果发现，直到
		 * 需要阻塞之前，继续从同一个队列读取要高效得多。
		 */
		fc_gatherstate->nextreader++;
		if (fc_gatherstate->nextreader >= fc_gatherstate->nreaders)
			fc_gatherstate->nextreader = 0;

		/* 我们是否访问过每个（存活的）TupleQueueReader? */
		fc_nvisited++;
		if (fc_nvisited >= fc_gatherstate->nreaders)
		{
			/*
			 * 如果（仍然）在本地运行计划，则返回 NULL，以便调用者可以
			 * 从计划的本地副本生成另一个元组。
			 */
			if (fc_gatherstate->need_to_scan_locally)
				return NULL;

			/* 没有什么可做的，除了等待事态发展。 */
			(void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
							 WAIT_EVENT_EXECUTE_GATHER);
			ResetLatch(MyLatch);
			fc_nvisited = 0;
		}
	}
}

/* ----------------------------------------------------------------
 *		ExecShutdownGatherWorkers
 *
 *		停止所有并行工作者。
 * ----------------------------------------------------------------
 */
static void fc_ExecShutdownGatherWorkers(GatherState *fc_node)
{
	if (fc_node->pei != NULL)
		ExecParallelFinish(fc_node->pei);

	/* 刷新读者数组的本地副本 */
	if (fc_node->reader)
		pfree(fc_node->reader);
	fc_node->reader = NULL;
}

/* ----------------------------------------------------------------
 *		ExecShutdownGather
 *
 *		销毁并行工作者的设置，包括并行上下文。
 * ----------------------------------------------------------------
 */
void ExecShutdownGather(GatherState *fc_node)
{
	fc_ExecShutdownGatherWorkers(fc_node);

	/* 现在销毁并行上下文。 */
	if (fc_node->pei != NULL)
	{
		ExecParallelCleanup(fc_node->pei);
		fc_node->pei = NULL;
	}
}

/* ----------------------------------------------------------------
 *						连接支持
 * ----------------------------------------------------------------
 */

/* ----------------------------------------------------------------
 *		ExecReScanGather
 *
 *		准备重新扫描 Gather 的结果。
 * ----------------------------------------------------------------
 */
void ExecReScanGather(GatherState *fc_node)
{
	Gather	   *fc_gather = (Gather *) fc_node->ps.plan;
	PlanState  *fc_outerPlan = outerPlanState(fc_node);

	/* 确保任何现有工作者都已优雅地关闭 */
	fc_ExecShutdownGatherWorkers(fc_node);

	/* 标记节点，以便在下次调用时重建共享状态 */
	fc_node->initialized = false;

	/*
	 * 设置子节点的 chgParam，告诉它下一个扫描可能在领导进程中
	 * 交付不同的行集。 （整体行集不应该改变，但领导进程的子集可能会；因此在这里和并行表扫描节点之间的节点
	 * 不得基于不变的行集假设进行优化。）
	 */
	if (fc_gather->rescan_param >= 0)
		fc_outerPlan->chgParam = bms_add_member(fc_outerPlan->chgParam,
											 fc_gather->rescan_param);

	/*
	 * 如果子节点的 chgParam 不是 null，则计划将由第一个 ExecProcNode 重新扫描。
	 * 注意：因为如果我们有 rescan_param，这样做不会产生任何结果，所以当前保证并行感知的子节点
	 * 在收到 ReInitializeDSM 调用之前不会看到 ReScan 调用。
	 * 这种顺序可能不是可靠的。 一个好的经验法则是 ReInitializeDSM 应该仅重置共享状态，ReScan
	 * 应仅重置本地状态，且任何依赖于这两个步骤完成的内容必须等到第一次 ExecProcNode 调用后才能执行。
	 */
	if (fc_outerPlan->chgParam == NULL)
		ExecReScan(fc_outerPlan);
}
